package com.atguigu.app.func;


import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.EduConfig;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.JedisUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private DruidDataSource druidDataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidDSUtil.getDruidDataSource();
    }


    //value:{"database":"gmall-221109-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,
    // "xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,
    // "spu_id":10,"order_id":987,"appraise":"1204",
    // "comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38",
    // "operate_time":null},"sink_table":"dim_comment_info","pk":"xx"}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        //redis
        String sinkTable = value.getString("sink_table");
        //{"id":1595211185799847960,"user_id":119,"nick_name":null}
        JSONObject data = value.getJSONObject("data");

        //redis
        //如果是更新操作
        if ("update".equals(value.getString("type"))){
            Jedis jedis = JedisUtil.getJedis();

            String redisKey = "DIM:"+ sinkTable.toUpperCase()+":"+data.getString("id");

            JSONObject jsonObject = new JSONObject();

            Set<Map.Entry<String, Object>> entries = data.entrySet();
            Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
            while (iterator.hasNext()){
                Map.Entry<String, Object> next = iterator.next();
                if (next == null){
                    jsonObject.put(next.getKey().toUpperCase(),next.getValue().toString());
                }else{
                    jsonObject.put(next.getKey().toUpperCase(),next.getValue());
                }
            }

            jedis.setex(redisKey,24 * 3600,jsonObject.toJSONString());

            jedis.close();
        }

        //获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();

        //获取sql upsert into db.tn(id,name,sex) values('1001','zs','male')
        String upsertSql = genSql(sinkTable, data);
        System.out.println("upsertSql--->" + upsertSql);

        //编译SQL
        PreparedStatement preparedStatement = connection.prepareStatement(upsertSql);

        //执行
        preparedStatement.execute();

        //提交
        connection.commit();

        //资源关闭
        preparedStatement.close();
        //归还
        connection.close();
    }


    /**
     * @param sinkTable 表名
     * @param data      {"id":"1001","name":"zs","sex":"male"}
     * @return upsert into db.tn(id,name,sex) values ('1001','zs','male')
     */
    private String genSql(String sinkTable, JSONObject data) {
        //取出列名
        Set<String> columns = data.keySet();

        //取出列值
        Collection<Object> values = data.values();
        System.out.println("upsert into " + EduConfig.HBASE_SCHEMA + "." + sinkTable + "(" + StringUtils.join(columns, ",") + ") values ('" +
                StringUtils.join(values, "','") + "')");


        return "upsert into " + EduConfig.HBASE_SCHEMA + "." + sinkTable + "(" +
                StringUtils.join(columns, ",") + ") values ('" +
                StringUtils.join(values, "','") + "')";

    }

    @Override
    public void close() throws Exception {
        druidDataSource.close();
    }
}
